// $Id: reactor.c 45 2011-04-16 08:39:25Z cedric.shih@gmail.com $
/*
 * Copyright (C) 2010 Cedric Shih <cedric.shih@gmail.com>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>

#include <event.h>

#include "log.h"
#include "socket.h"

#include "reactor.h"

#define nioreactor_lock(x) do { \
	if (x->radiator) { \
		niotrace("locking: %p", &x->lock); \
		if ((errno = pthread_mutex_lock(&x->lock)) < 0) { \
			nioerror("failed to lock: %s", strerror(errno)); \
			return NIO_ELOCK; \
		} \
	} \
} while (0)

#define nioreactor_unlock(x) do { \
	if (x->radiator) { \
		niotrace("unlocking: %p", &x->lock); \
		if ((errno = pthread_mutex_unlock(&x->lock)) < 0) { \
			nioerror("failed to unlock: %s", strerror(errno)); \
			return NIO_ELOCK; \
		} \
	} \
} while (0)

struct nioreactor *nioreactor_new(struct event_base *selector)
{
	struct nioreactor *reactor;

	niodebug("constructing");

	if (!(reactor = calloc(1, sizeof(struct nioreactor)))) {
		nioerror("failed to calloc nioreactor");
		goto error;
	}

	TAILQ_INIT(&reactor->futures);
	reactor->evbase = selector;

	return reactor;

error:
	nioreactor_free(reactor);
	return NULL;
}

void nioreactor_free(struct nioreactor *reactor)
{
	if (!reactor) {
		return;
	}

	if (reactor->pipe[PIPERECV] > 0) {
		close(reactor->pipe[PIPERECV]);
	}

	if (reactor->pipe[PIPESEND] > 0) {
		close(reactor->pipe[PIPESEND]);
	}

	free(reactor);
}

int nioreactor_insert(struct nioreactor *reactor, struct niofuture *future,
		struct niosocket *socket, enum niofuture_flag flag,
		struct timeval *timeout, niofuture_cb cb, void *cbarg)
{
	int rc;

	niodebug("inserting: %p", future);

	nioreactor_lock(reactor);

	if (future->flags & NIO_PENDING) {
		nioerror("pending in queue");
		rc = NIO_EEXIST;
		goto finally;
	} else if (event_initialized(&future->event)) {
		if (event_pending(&future->event, EV_READ, NULL)) {
			nioerror("pending read");
			rc = NIO_EEXIST;
			goto finally;
		} else if (event_pending(&future->event, EV_WRITE, NULL)) {
			nioerror("pending write");
			rc = NIO_EEXIST;
			goto finally;
		} else if (event_pending(&future->event, EV_TIMEOUT, NULL)) {
			nioerror("pending timeout");
			rc = NIO_EEXIST;
			goto finally;
		}
	}

	future->socket = socket;
	future->cb = cb;
	future->cbarg = cbarg;
	future->flags = flag;

	if (timeout != NULL) {
		memcpy(&future->timeout, timeout, sizeof(future->timeout));
		future->flags |= NIO_TIMEUP;
	}

	if (reactor->radiator) {
		if ((rc = nioreactor_wake_up(reactor))) {
			nioerror("failed to wake up reactor");
			rc = NIO_ELOCK;
			goto finally;
		}

		niodebug("adding to queue: %p", future);

		TAILQ_INSERT_TAIL(&reactor->futures, future, next);
		future->flags |= NIO_PENDING;
	} else {
		if ((rc = nioreactor_activate(reactor, future))) {
			nioerror("failed to activate: %p", future);
			rc = NIO_EUNKNOWN;
			goto finally;
		}
	}

	rc = 0;

finally:
	nioreactor_unlock(reactor);
	return rc;
}

int nioreactor_loop(struct nioreactor *reactor,
		struct nioradiator *radiator)
{
	int rc;

	niodebug("starting loop");

	if (radiator) {
		if ((rc = pipe(reactor->pipe)) < 0) {
			rc = errno ? errno : -1;
			nioerror("failed to create pipe: %s", strerror(errno));
			rc = NIO_ELOCK;
			goto finally;
		}

		event_set(&reactor->event, reactor->pipe[PIPERECV], EV_READ,
				nioreactor_awaken_cb, reactor);

		if ((rc = event_base_set(reactor->evbase, &reactor->event))) {
			nioerror("failed to assign event base");
			rc = NIO_ELOCK;
			goto finally;
		}

		if ((rc = event_add(&reactor->event, NULL))) {
			nioerror("failed to add event");
			rc = NIO_ELOCK;
			goto finally;
		}

		if ((rc = pthread_mutex_init(&reactor->lock, NULL))) {
			nioerror("failed to init mutex: %s", strerror(rc));
			rc = NIO_ELOCK;
			goto finally;
		}

		reactor->breaking = 0;
		reactor->radiator = radiator;
	}

	if ((rc = event_base_dispatch(reactor->evbase))) {
		if (rc != 1) {
			nioerror("failed to dispatch event base");
			rc = NIO_EUNKNOWN;
			goto finally;
		}
	}

	niodebug("exiting loop");
	rc = 0;

finally:
	if (radiator) {
		niodebug("destroying mutex");

		if ((rc = pthread_mutex_destroy(&reactor->lock))) {
			nioerror("failed to destroy mutex: %s", strerror(rc));
		}

		reactor->radiator = NULL;
	}
	return rc;
}

int nioreactor_break(struct nioreactor *reactor)
{
	int rc;

	niodebug("breaking loop");

	nioreactor_lock(reactor);

	if (reactor->radiator) {
		reactor->breaking = 1;

		if ((rc = nioreactor_wake_up(reactor))) {
			nioerror("failed to wake up");
			rc = NIO_ELOCK;
			goto finally;
		}
	}

//	if ((rc = event_base_loopbreak(reactor->evbase))) {
//		nioerror("failed to break event base loop");
//		rc = NIO_EUNKNOWN;
//		goto finally;
//	}

	rc = 0;

finally:
	nioreactor_unlock(reactor);
	return rc;
}

inline struct event_base *nioreactor_get_selector(struct nioreactor *reactor)
{
	return reactor->evbase;
}

void nioreactor_event_cb(int fd, short event, void *arg)
{
	struct niofuture *future = arg;

	if (!(event & EV_TIMEOUT)) {
		future->flags &= (~NIO_TIMEUP);
	}

	nioreactor_dispatch(future->reactor, future);
}

int nioreactor_dispatch(struct nioreactor *reactor, struct niofuture *future)
{
	int rc;

	nioreactor_lock(reactor);

	if (reactor->radiator) {
		if ((rc = nioradiator_dispatch(
				reactor->radiator, future))) {
			nioerror("failed to dispatch with radiator");
			goto finally;
		}
	} else {
		niodebug("invoking: %p", future->cb);
		(*future->cb)(future, future->socket,
				future->flags, future->cbarg);
	}

	rc = 0;

finally:
	nioreactor_unlock(reactor);
	return rc;
}

int nioreactor_contains(struct nioreactor *reactor, struct niofuture *future,
		int *answer)
{
	int rc;

	nioreactor_lock(reactor);

	if (!reactor || !future || !answer) {
		nioerror("null argument(s)");
		rc = NIO_EINVAL;
		goto finally;
	}

	if ((future->flags & NIO_PENDING) || (event_initialized(&future->event) &&
			event_pending(&future->event, EV_READ|EV_WRITE, NULL))) {
		*answer = 1;
	} else {
		*answer = 0;
	}

	rc = 0;

finally:
	nioreactor_unlock(reactor);
	return rc;
}

int nioreactor_remove(struct nioreactor *reactor, struct niofuture *future)
{
	int rc;

	niodebug("removing event: %p", future);

	nioreactor_lock(reactor);

	if (future->flags & NIO_PENDING) {
		niodebug("removing from queue: %p", future);

		TAILQ_REMOVE(&reactor->futures, future, next);
		future->flags &= (~NIO_PENDING);

	} else if (event_initialized(&future->event) &&
			event_pending(&future->event,
			EV_READ|EV_WRITE|EV_TIMEOUT, NULL)) {
		niodebug("deleting event");

		event_del(&future->event);
	} else {
		niodebug("not pending");
		rc = NIO_ENOENT;
		goto finally;
	}

	rc = 0;

finally:
	nioreactor_unlock(reactor);
	return rc;
}

void nioreactor_awaken_cb(int fd, short event, void *arg)
{
	char buffer[32];

	while(read(fd, buffer, sizeof(buffer)) == sizeof(buffer));

	nioreactor_awaken_do(arg);
}

int nioreactor_awaken_do(struct nioreactor *reactor)
{
	int rc;
	struct niofuture *future;

	niodebug("starting awaken routine");

	nioreactor_lock(reactor);

	if (reactor->breaking) {
		niodebug("requested to break");

//		reactor->radiator = NULL;

//		niodebug("unlocking mutex");
//
//		if ((rc = pthread_mutex_unlock(&reactor->lock)) < 0) {
//			nioerror("failed to unlock: %s", strerror(rc));
//			rc = NIO_ELOCK;
//			goto finally;
//		}

		niodebug("deleting wake up event");

		if ((rc = event_del(&reactor->event))) {
			nioerror("failed to delete event");
			rc = NIO_ELOCK;
			goto finally;
		}

		niodebug("closing wake up pipe");

		if ((rc = close(reactor->pipe[0]))) {
			nioerror("failed to close pipe: %s", strerror(rc));
			rc = NIO_ELOCK;
			goto finally;
		}

		if ((rc = close(reactor->pipe[1]))) {
			nioerror("failed to close pipe: %s", strerror(rc));
			rc = NIO_ELOCK;
			goto finally;
		}
	} else {
		niodebug("processing events");

		while ((future = TAILQ_FIRST(&reactor->futures))) {
			niodebug("removing from queue: %p", future);

			TAILQ_REMOVE(&reactor->futures, future, next);
			future->flags &= (~NIO_PENDING);

			if ((rc = nioreactor_activate(reactor, future))) {
				nioerror("failed to activate: %p", future);
				goto finally;
			}
		}

		if ((rc = event_add(&reactor->event, NULL))) {
			nioerror("failed to add event");
			goto finally;
		}
	}

	rc = 0;

finally:
	nioreactor_unlock(reactor);
	return rc;
}

int nioreactor_activate(struct nioreactor *reactor, struct niofuture *future)
{
	int rc;
	struct timeval *tv;

	niodebug("activating: %p", future);

	if (future->flags & NIO_READABLE) {
		niotrace("setting event for read: %p", &future->event);
		event_set(&future->event, future->socket->descriptor,
				EV_READ, nioreactor_event_cb, future);
	} else if (future->flags & NIO_WRITABLE) {
		niotrace("setting event for write: %p", &future->event);
		event_set(&future->event, future->socket->descriptor,
				EV_WRITE, nioreactor_event_cb, future);
	} else {
		niotrace("setting event for timer: %p", &future->event);
		evtimer_set(&future->event, nioreactor_event_cb, future);
	}

	niodebug("setting event base: %p", &future->event);

	if ((rc = event_base_set(reactor->evbase, &future->event))) {
		nioerror("failed to assign event base");
		goto finally;
	}

	if (future->flags & NIO_TIMEUP) {
		niodebug("timeout in %ld.%06ld second",
				future->timeout.tv_sec,
				future->timeout.tv_usec);
		tv = &future->timeout;
	} else {
		tv = NULL;
	}

	niotrace("adding event: %p", &future->event);

	if ((rc = event_add(&future->event, tv))) {
		nioerror("failed to add event");
		goto finally;
	}

finally:
	return rc;
}

int nioreactor_wake_up(struct nioreactor *reactor)
{
	niodebug("waking up");
	return write(reactor->pipe[PIPESEND], "", 1) < 0;
}
